package org.example.process;


import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

public class EventTimeTimerExample {

    public static void main(String[] args) throws Exception {
        // 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 模拟输入数据流
        DataStream<Event> stream = env.fromElements(
                new Event("key1", "event1", 1000L),
                new Event("key1", "event2", 3000L)
        ).assignTimestampsAndWatermarks(WatermarkStrategy.forMonotonousTimestamps());

        // 使用 KeyedProcessFunction 处理事件
        stream
                .keyBy(Event::getKey)
                .process(new MyProcessFunction())
                .print();

        // 执行程序
        env.execute("Event Time Timer Example");
    }

    // 自定义 ProcessFunction
    public static class MyProcessFunction extends KeyedProcessFunction<String, Event, String> {

        @Override
        public void processElement(Event value, Context ctx, Collector<String> out) throws Exception {
            // 注册一个事件时间定时器，延迟 5 秒触发
            long timerTimestamp = value.getTimestamp() + 5000L;
            ctx.timerService().registerEventTimeTimer(timerTimestamp);
            System.out.println("processElement 当前时间: " + System.currentTimeMillis() + ",水位线 " +
                    ctx.timerService().currentWatermark());
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
            // 定时器触发时输出日志
            out.collect("Timer triggered for key: " + ctx.getCurrentKey()
                    + " at timestamp: " + timestamp + " at currentWatermark " +  ctx.timerService().currentWatermark());
            System.out.println("onTimer 当前时间: " + System.currentTimeMillis());

        }
    }

    // 数据模型
    public static class Event {
        public String key;
        public String value;
        public long timestamp;

        public Event(String key, String value, long timestamp) {
            this.key = key;
            this.value = value;
            this.timestamp = timestamp;
        }

        public String getKey() {
            return key;
        }

        public long getTimestamp() {
            return timestamp;
        }
    }
}
